package com.offcn.bigdata.spark.p2

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 学习Spark RDD的transformation操作
  *     flatMap
  *     map
  *     filter
  *     -----------------------
  *     mapPartitions & mapPartitionsWithIndex
  *     sample
  *     union
  *     distinct
  *     groupByKey&groupBy
  *     join
  *     reduceByKey & foldByKey
  *     repartition & coalesce
  *     combineByKey & aggregateByKey
  *
  *     sortByKey
  *
  */
object _01SparkTransformationOps {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf()
                    .setMaster("local[*]")
                    .setAppName(s"${_01SparkTransformationOps.getClass.getSimpleName}")
        val sc = new SparkContext(conf)

//        mapPartitionsOps(sc)
//        sampleOps(sc)
//        unionOps(sc)
//        groupByKeyOps(sc)
//        joinOps(sc)
//        rbkAndfbkOps(sc)
        repartitionOps(sc)
        sc.stop()
    }

    /**
      * 重分区：
      *     原先有N个分区，经过重分区算子之后，变为了M个分区，那么M可能大于N，也可能小于N。
      *     coalesce(partition, shuffle = false)    ： 通常用作分区减小的操作，将原先的分区减少到指定个数的分区
      *             通常分区前后的分区个数存在整数倍的关系。是一个窄依赖操作。前后的分区对应关系是1：N
      *     repartition ：coalesce(partition, shuffle = true)，是一个宽依赖的操作
      *             hash分区
      * @param sc
      */
    def repartitionOps(sc: SparkContext): Unit = {
        val rdd = sc.parallelize(1 to 100000, 10)
        println("-----------------------过滤前---------------------------")
        rdd.mapPartitionsWithIndex((index, it) => {
            val list = it.toList
            println(s"rdd{${index}}分区中的的数据个数：${list.size}")
            list.map(num => num).toIterator
        }).count()
        println("-----------------------过滤后---------------------------")
        val filtered = rdd.filter(num => num % 3 != 0)//过滤掉了1/3的数据
        filtered.mapPartitionsWithIndex((index, it) => {
            val list = it.toList
            println(s"过滤之后filtered{${index}}分区中的的数据个数：${list.size}")
            list.map(num => num).toIterator
        }).count()
        println("-----------------------分区合并(减小)---------------------------")
        val coalesceRDD = filtered.coalesce(5)
        coalesceRDD.mapPartitionsWithIndex((index, it) => {
            val list = it.toList
            println(s"减少分区之后rdd分区{${index}}中的的数据个数：${list.size}")
            list.map(num => num).toIterator
        }).count()
        println("-----------------------分区合并(增大)---------------------------")
        val repartioned = filtered
                    .repartition(15)
                    //coalesce(15, shuffle = true)
        repartioned.mapPartitionsWithIndex((index, it) => {
            val list = it.toList
            println(s"增大分区之后rdd分区{${index}}中的的数据个数：${list.size}")
            list.map(num => num).toIterator
        }).count()
    }

    /**
      * reduceByKey和foldByKey
      * reduceByKey((A1, A2) => A3)
      * foldByKey(zero)((A1, A2) => A3)
      * 关系相当于scala集合中的reduce和fold的关系
      */
    def rbkAndfbkOps(sc: SparkContext): Unit = {
        val lines = sc.parallelize(List(
            "hello moto",
            "hello nokia",
            "hello xiaomi",
            "hello huawei",
            "shit sangsumg"
        ))
        val pairs:RDD[(String, Int)] = lines.flatMap(_.split("\\s+")).map((_, 1))

        pairs.reduceByKey((v1, v2) => v1 + v2).foreach(println)
        println("--------------------------------------------------------")
        pairs.foldByKey(0)(_+_).foreach(println)

    }

    /**
      * join(otherRDD): 和otherRDD进行join操作，类似于java中inner join
      *     要求该rdd和other都是k-v，也就是说通过k作为外键进行管理
      * 关联操作：
      *    A表(id)和B(aid) id和aid为两张表的关联字段
      *     交叉连接
      *         xxxx A a [across] join B b
      *         没有关联条件,也就意味着有笛卡尔积，不用
      *     内连接
      *         xxx A a inner join B b on a.id = b.aid
      *         xxx A a, B b where a.id = b.aid
      *         返回这两张表中的交集部分
      *     外连接
      *         左外连接：xxx A a left outer join B b on a.id = b.aid
      *             返回左表所有和右表关联上的结果，没有关联上的显示为null
      *         右外连接：xxx A a right outer join B b on a.id = b.aid
      *             和做外连接相反
      *     全连接
      *         xxx A a full outer join B b on a.id = b.aid
      *         左外连接+右外连接
      * @param sc
      */
    def joinOps(sc: SparkContext): Unit = {
        case class Student(id: Int, name: String, age: Int, gender: Int)
        case class Score(sid: Int, course: String, score: Double)
        val stus = sc.parallelize(List(
            Student(1, "林博", 18, 0),
            Student(2, "单松", 19, 1),
            Student(3, "张皓", 20, 0),
            Student(4, "王建", 20, 1),
            Student(106, "冯岩", 30, 1)
        ))
        val scores = sc.parallelize(List(
            Score(2, "语文", 80),
            Score(3, "数学", 70.5),
            Score(1, "英语", 30.5),
            Score(4, "体育", 60),
            Score(105, "信息技术", 100)
        ))

        // 关联学习信息及其考试成绩
        val id2Stu:RDD[(Int, Student)] = stus.map(stu => (stu.id, stu))
        val id2Score:RDD[(Int, Score)] = scores.map(score => (score.sid, score))
        println("------------join结果--------------")
        val joinedRDD:RDD[(Int, (Student, Score))] = id2Stu.join(id2Score)
//        joinedRDD.foreach(t => {
//            println(s"学生id：${t._1}, name: ${t._2._1.name}")
//        })
        joinedRDD.foreach{case (id, (stu, score)) => {
            println(s"学生id：${id}, name: ${stu.name}, 性别：${stu.gender}, 科目：${score.course}, 成绩: ${score.score}")
        }}
        println("------------left join结果--------------")
        //显示所有学生信息
        val leftJoinedRDD:RDD[(Int, (Student, Option[Score]))] = id2Stu.leftOuterJoin(id2Score)
        leftJoinedRDD.foreach{case (id, (stu, scoreOption)) => {
            println(s"学生id：${id}, name: ${stu.name}, 性别：${stu.gender}, 成绩：${scoreOption.getOrElse(null)}")
        }}
        println("------------right join结果--------------")
        //显示所有成绩信息
        val rightJoinedRDD:RDD[(Int, (Option[Student], Score))] = id2Stu.rightOuterJoin(id2Score)
        rightJoinedRDD.foreach{case (id, (stuOption, score)) => {
            println(s"学生id：${id}, 学生信息: ${stuOption.getOrElse(null)}, 成绩：${score}")
        }}
        println("------------full join结果--------------")
        //显示所有信息
        val fullJoinedRDD:RDD[(Int, (Option[Student], Option[Score]))] = id2Stu.fullOuterJoin(id2Score)
        fullJoinedRDD.foreach{case (id, (stuOption, scoreOption)) => {
            println(s"学生id：${id}, 学生信息: ${stuOption.getOrElse(null)}, 成绩：${scoreOption.getOrElse(null)}")
        }}
    }

    /**
      * groupByKey[K](parallelize): RDD[(K, Iterable[V])]: 使用多少个并行度来执行按照key进行分组，所操作的就是k-v
      *
      * @param sc
      */
    def groupByKeyOps(sc: SparkContext): Unit = {
        case class Student(name: String, age: Int, gender: Int)
        val stus = sc.parallelize(List(
            Student("郭雪磊", 18, 0),
            Student("李佳奥", 19, 1),
            Student("元永劫", 20, 0),
            Student("许迎港", 20, 1)
        ))
        //按照性别进行负责
        val gender2Stu:RDD[(Int, Iterable[Student])] = stus.map(stu => (stu.gender, stu))
            .groupByKey()

        gender2Stu.foreach{case (gender, stus) => {
            println(s"性别为: ${if(gender == 0) '女' else '男'} 对应的学生有：${stus.toList}")
        }}
        println("---------------groupBy---------------------")

        stus.groupBy(stu => stu.gender).foreach{case (gender, stus) => {
            println(s"性别为: ${if(gender == 0) '女' else '男'} 对应的学生有：${stus.toList}")
        }}
    }

    /**
      * union: 联合，相当于sql中的union all
      * distinct: 去除重复操作
      */
    def unionOps(sc: SparkContext): Unit = {
        val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5), 2)
        println("----------------rdd1中的数据------------------")
        rdd1.mapPartitionsWithIndex((index, it) => {
            val list = it.toList
            println(s"rdd1分区{${index}}的数据：${list.mkString("[", ", ", "]")}")
            list.map(_ * 2).toIterator
        }).count()
        val rdd2 = sc.parallelize(List(1, 9, 6, 4, 5), 2)
        println("----------------rdd2中的数据------------------")
        rdd2.mapPartitionsWithIndex((index, it) => {
            val list = it.toList
            println(s"rdd2分区{${index}}的数据：${list.mkString("[", ", ", "]")}")
            list.map(_ * 2).toIterator
        }).count()
        val unionRDD = rdd1.union(rdd2)//窄依赖操作
        println("----------------union操作-------------------")
        unionRDD.mapPartitionsWithIndex((index, it) => {
            val list = it.toList
            println(s"unionRDD分区{${index}}的数据：${list.mkString("[", ", ", "]")}")
            list.map(_ * 2).toIterator
        }).count()
        println("----------------distinct去重操作-------------")
        val distincted = unionRDD.distinct()//宽依赖操作
        distincted.mapPartitionsWithIndex((index, it) => {
            val list = it.toList
            println(s"distincted分区{${index}}的数据：${list.mkString("[", ", ", "]")}")
            list.map(_ * 2).toIterator
        }).count()
    }

    /**
      * sample(withReplacement: Boolean, fraction: Double, seed: Long): 抽样，采样
      *     withReplacement : 采样的方式，是否放回的抽样，true为放回，false为不放回
      *     fraction        : 采样的比例，0~1之间的double
      *     seed            : 抽样的随机数种子，有默认值，我们不需要进行传递
      *  注意：该算子为非准确抽样，也就是说按照比例抽取的结果不一定准确，可能在上下浮动，但是对整体影响不大。
      *  该算子通常在用于处理数据倾斜的时候比较有用。
      */
    def sampleOps(sc: SparkContext): Unit = {
        val listRDD:RDD[Int] = sc.parallelize(1 to 1000000)
        //使用无放回的抽样方式，抽取10%的数据
        var sampledRDD:RDD[Int] = listRDD.sample(false, 0.1)
        println(s"无放回RDD采样数据个数为: ${sampledRDD.count()}")
        //使用有放回的抽样方式，抽取10%的数据
        sampledRDD = listRDD.sample(true, 0.1)
        println(s"有放回RDD采样数据个数为: ${sampledRDD.count()}")
    }

    /**
      * map(p: (A) => B)
      * mapPartitions(p: (Iterator[A]) => Iterator[B])
      *  mapPartitions就是map的批量操作的版本，每次要处理的是一个partition中的数据，而map每次处理的是一条记录。
      */
    def mapPartitionsOps(sc: SparkContext): Unit = {
        //通过并行化的集合来构建rdd
        val rdd:RDD[Int] = sc.parallelize(1 to 10)
        println("----------------map的操作----------------")
        rdd.map(num => {
            print(num + "\t")
            num * 2
        }).count()
        println("\r\n----------------mapPartitions的操作----------------")
        rdd.mapPartitions(it => {
            val list = it.toList
            println(s"分区中的数据：${list.mkString("[", ", ", "]")}")
            list.map(_ * 2).toIterator
        }).count()
        println("----------------mapPartitionsWithIndex的操作----------------")
        rdd.mapPartitionsWithIndex((index, it) => {
            val list = it.toList
            println(s"分区{${index}}中的数据：${list.mkString("[", ", ", "]")}")
            list.map(_ * 2).toIterator
        }).count()
    }
}
